package com.huawei.codearts.service;

import io.etcd.jetcd.*;
import io.etcd.jetcd.watch.WatchResponse;
import io.etcd.jetcd.kv.DeleteResponse;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.kv.PutResponse;
import io.etcd.jetcd.lease.LeaseGrantResponse;
import io.etcd.jetcd.lock.LockResponse;
import io.etcd.jetcd.watch.WatchEvent;
import io.grpc.stub.StreamObserver;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Scanner;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
@RequiredArgsConstructor
public class EtcdService {

    private final Client etcdClient;  // 通过 @Configuration 方式配置对象
//    private final Client etcdClient = EtcdUtils.etcdClient();  // 通过调用的方式获取对象

    /** put操作 */
    public Object put(String name, String value) {

        ByteSequence keyByte = ByteSequence.from(name, StandardCharsets.UTF_8);
        ByteSequence valueByte = ByteSequence.from(value, StandardCharsets.UTF_8);
//        CompletableFuture<PutResponse> future = etcdClient.getKVClient().put(keyByte, valueByte);  // 异步
        PutResponse response = etcdClient.getKVClient().put(keyByte, valueByte).join();  // 同步
        System.out.println("Put response = " + response);
//
//        try {
////            future.get(); // 阻塞并等待结果
//            System.out.println("future.get() = " + future.get());
//        } catch (InterruptedException | ExecutionException e) {
//            // 处理异常
//        }

//        future.thenAccept(result -> {
//            // 操作完成时的回调
//            System.out.println("result = " + result);
//        });

        // 自定义map
        HashMap<String, Object> dataMap = new HashMap<>();
        dataMap.put("data", response);
        dataMap.put("header", response.getHeader());

        return dataMap;
    }

    /** get操作 */
    public Object get(String name) {

//        ByteSequence keyByte = ByteSequence.from(name, StandardCharsets.UTF_8);
        ByteSequence keyByte = ByteSequence.from(name, Charset.defaultCharset());
        GetResponse getResponse = etcdClient.getKVClient().get(keyByte).join();
        System.out.println("Get response [" + name + "] = " + getResponse.getKvs().get(0).getValue().toString(Charset.defaultCharset()));

        // 自定义map
        HashMap<String, Object> dataMap = new HashMap<>();
        dataMap.put("key", name);
        dataMap.put("value", getResponse.getKvs().get(0).getValue().toString(Charset.defaultCharset()));  // key的值
//        dataMap.put("data", getResponse);

        return dataMap;
    }

    /** delete操作 */
    public Object delete(String name) {

        ByteSequence keyByte = ByteSequence.from(name, Charset.defaultCharset());
        DeleteResponse response = etcdClient.getKVClient().delete(keyByte).join();
        System.out.println("Delete response = " + response);

        // 自定义map
        HashMap<String, Object> dataMap = new HashMap<>();
        dataMap.put("data", response);
        dataMap.put("delete", response.getDeleted());  // response.getDeleted() == 1 删除成功 0失败

        return dataMap;
    }

    /** watch 监听操作 */
    public Object watch() {

        // create watchClient
        Watch watchClient = etcdClient.getWatchClient();
        ByteSequence key = ByteSequence.from("/config/application".getBytes());
        try {
            watchClient.watch(key, new Watch.Listener() {
                @Override
                public void onNext(WatchResponse response) {
                    for (WatchEvent event : response.getEvents()) {
                        System.out.println("操作: " + event.getEventType());
                        KeyValue kv = event.getKeyValue();
                        System.out.println("当前kv: ");
                        System.out.println(new String(kv.getKey().getBytes()));
                        System.out.println(new String(kv.getValue().getBytes()));
                        System.out.println(kv.getVersion());
                        System.out.println(kv.getCreateRevision());
                        System.out.println(kv.getModRevision());
                        System.out.println(kv.getLease());
                    }
                }

                @Override
                public void onError(Throwable throwable) {
                    log.info("停止监听... error: {}", throwable.getMessage());
                }

                @Override
                public void onCompleted() {
                    log.info("结束监听...");
                }
            });
            System.out.println("键盘输入结束监听!");
            Scanner sc = new Scanner(System.in);
            sc.nextLine();
        } finally {
            if (watchClient != null) {
                watchClient.close();
            }
        }

        // 自定义map
        HashMap<String, Object> dataMap = new HashMap<>();
        return dataMap;
    }

    /** lock、unlock等操作 */
    public Object lock() {

        // 初始化etcd客户端
        Thread currentThread = Thread.currentThread();
        Lock lockClient = etcdClient.getLockClient();
        Lease leaseClient = etcdClient.getLeaseClient();

        String lockPath;
        long leaseId;

        // 尝试获取锁
        String lockKey = "/my_lock";
        long leaseAcquireTime = 10; // 租约时间（秒）
        try {
            // 创建一个租约
//            LeaseGrantResponse ttl = leaseClient.grant(TimeUnit.NANOSECONDS.toSeconds(leaseAcquireTime)).get();
            LeaseGrantResponse ttl = leaseClient.grant(leaseAcquireTime).get();
            leaseId = ttl.getID();
            log.info("创建一个租约...ID: {}, ttl: {}", leaseId, ttl.getTTL());

            // 自动续约-防止ttl超时线程未执行完
            leaseClient.keepAlive(leaseId, new StreamObserver() {
                @Override
                public void onNext(Object value) {
                    //执行续约之后的回调
                    log.info("执行续约...ID: {}, ttl: {}", leaseId, ttl.getTTL());
                }

                @Override
                public void onError(Throwable t) {
                    //异常，就会停止续约，比如调用revoke取消续约，租约不存在
                    log.info("停止续约...{} 原因: {}", leaseId, t.getMessage());
                }

                @Override
                public void onCompleted() {
                    log.info("onCompleted...");
                }
            });

            // 使用etcd的分布式锁API
//            LockResponse lockResponse = lockClient.lock(ByteSequence.from(lockKey.getBytes()), leaseId).get(lockAcquireTime, TimeUnit.SECONDS);
            LockResponse lockResponse = lockClient.lock(ByteSequence.from(lockKey.getBytes()), leaseId).get();
            if (lockResponse != null) {
                long start = System.currentTimeMillis();
                lockPath = lockResponse.getKey().toString(StandardCharsets.UTF_8);
                log.info("线程：{} 加锁成功，锁路径：{}", currentThread.getName(), lockPath);

                // 模拟线程执行时长
                TimeUnit.SECONDS.sleep(15);
                log.info("执行时长... {} ms", System.currentTimeMillis() - start);

                // 删除租约
                if (leaseId != 0L) {
                    leaseClient.revoke(leaseId);
                    log.info("leaseId revoke leaseId... {}", leaseId);
                }
                // 释放锁
                if (lockPath != null) {
                    lockClient.unlock(ByteSequence.from(lockPath.getBytes())).get();
                    log.info("Lock released leaseId... {}", leaseId);
                }
            } else {
                log.info("Failed to acquire lock...");
            }
        }catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        } finally {
            // 关闭客户端
//            etcdClient.close();
        }


        // 模拟不同线程
        //name-锁标识
        ByteSequence name = ByteSequence.from("mutex1".getBytes());
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                try {
                    //创建一个租约
                    LeaseGrantResponse ttl = leaseClient.grant(10).get();
                    log.info("创建租约...ID: {}, ttl: {}", ttl.getID(), ttl.getTTL());
                    //自动续约-防止ttl超时线程未执行完
                    leaseClient.keepAlive(ttl.getID(), new StreamObserver() {
                        @Override
                        public void onNext(Object value) {
                            //执行续约之后的回调
                            log.info("执行续约...ID: {}, ttl: {}", ttl.getID(), ttl.getTTL());
                        }

                        @Override
                        public void onError(Throwable t) {
                            //异常，就会停止续约，比如调用revoke取消续约，租约不存在
                            log.info("停止续约...{} 原因: {}", ttl.getID(), t.getMessage());
                        }

                        @Override
                        public void onCompleted() {
                            log.info("onCompleted...");
                        }
                    });
                    log.info("尝试获取锁...");
                    //尝试获取锁
                    ByteSequence key = lockClient.lock(name, ttl.getID()).get().getKey();
                    long start = System.currentTimeMillis();
                    log.info("获得锁...key: {}", new String(key.getBytes()));
                    //模拟线程执行时长
                    TimeUnit.SECONDS.sleep(3);
                    log.info("执行时长... {} ms", System.currentTimeMillis() - start);
                    //取消租约
                    leaseClient.revoke(ttl.getID());
                    log.info("取消租约...{}", ttl.getID());
                    lockClient.unlock(key);
                    log.info("释放锁...");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }, "线程" + i).start();
        }


        // 自定义map
        HashMap<String, Object> dataMap = new HashMap<>();
        return dataMap;
    }

}
